package com.nl.pulsardemo.io.coutom;

import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;

/**
 * @author : ZhaoY
 * @description :
 * @date : 2021-02-25 15:50
 **/
public class TestSink implements Sink<String> {
    private static final String FILENAME = "/tmp/test-out";
    private BufferedWriter bw = null;
    private FileWriter fw = null;

    @Override
    public void open(Map<String, Object> map, SinkContext sinkContext) throws Exception {
        File file = new File(FILENAME);
        // if file doesnt exists, then create it
        if (!file.exists()) {
            file.createNewFile();
        }
        fw = new FileWriter(file.getAbsoluteFile(), true);
        bw = new BufferedWriter(fw);
    }

    @Override
    public void write(Record<String> record) throws Exception {
        try {
            bw.write(record.getValue());
            bw.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() throws Exception {
        try {
            if (bw != null)
                bw.close();
            if (fw != null)
                fw.close();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}
